package ldh.im.fxgui.service;

import ldh.im.ImConfig;
import ldh.im.dto.base.OkListReq;
import ldh.im.dto.base.OkListResp;
import ldh.im.dto.base.OkResp;
import ldh.im.dto.business.base.*;
import ldh.im.dto.enums.*;
import ldh.im.dto.business.group.GroupCreateReq;
import ldh.im.dto.business.group.GroupInfoListReq;
import ldh.im.dto.business.message.*;
import ldh.im.dto.business.item.*;
import ldh.im.dto.business.user.UserInfoListReq;
import ldh.im.fxbase.dao.*;
import ldh.im.fxbase.data.pojo.*;
import ldh.im.fxbase.data.pojo.status.*;
import ldh.im.fxbase.data.pojo.ImMessageActionData;
import ldh.im.fxbase.rpc.service.RpcBaseService;
import ldh.im.fxbase.rpc.service.RpcGroupService;
import ldh.im.fxbase.service.*;
import ldh.im.fxbase.util.*;
import ldh.im.fxbase.rpc.service.RpcMessageService;
import ldh.im.fxbase.rpc.service.RpcPullMessageService;
import ldh.im.fxgui.util.UiUtil;
import ldh.im.util.ConfigUtil;
import ldh.im.util.DateUtil;
import ldh.rpc.TerminalType;
import ldh.rpc.cache.CacheService;
import ldh.rpc.util.JsonUtil;
import org.slf4j.Logger;


import java.sql.SQLException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * 数据同步服务：
 * 拉取部门，拉取用户，拉取群，发送未成功的消息，发送未成功的消息动作，拉取会话
 * 拉取消息，拉取消息动作
 */
public class SynDataService {

    private static final Logger LOGGER = LoggerUtil.getFileLogger(SynDataService.class);

    public SynDataService() {}

    public void synData(String userId) {
        long startTime = System.currentTimeMillis();
        int size = 0;
        try {
            size += synCommonData(userId);
            size += synMessages(userId);            // 同步消息
            size += synMessageAction(userId);       // 同步消息事件
            size += synLabels(userId);              // 同步标签
            size += synCollects(userId);            // 同步收藏
            size += synFeedbacks(userId);           // 同步反馈
            size += synSensitiveWords(userId);      // 同步敏感词
            LOGGER.info("synData end!!!!!!!!!!!!!!");
        } catch (Exception e) {
            LOGGER.error("同步数据错误", e);
        } finally {
            long time = System.currentTimeMillis() - startTime;
            LOGGER.info("synData total time: {}, total size: {}", time, size);
        }
    }

    // 同步数据：机构，群，用户
    public int synCommonData(String userId) throws Exception {
        int size = 0;
        size += synConfigs(userId);                  // 同步配置
        size += synDepartments();                    // 同步机构
        size += synUsers();                          // 同步用户
        size += synGroups(userId);                   // 同步群
        size += synModules(userId);                  // 同步模块信息
        size += sendNeedGroups(userId);              // 发送未成功的群
        size += sendNeedMessages(userId);            // 发送未成功的消息
        size += sendMessageActionStatus(userId);     // 发送未同步成功的消息变更事件
        size += synImSessions(userId);               // 同步消息会话
        return size;
    }

    /**
     * 同步消息
     * @param userId
     */
    public int synMessages(String userId) throws Exception {
        ImSessionDao imSessionDao = BeanFactory.getInstance().getBean(ImSessionDao.class);
        return synData(userId, ImSettingName.SynMessage.name(), (localDate, isLatestChange) -> {
            int size = 0;
            long time = System.currentTimeMillis();
            int start = 0;
            LocalDateTime startDay = localDate;
            int pullPreDay = ConfigUtil.getInt("message.pull.preDay", -30);
            LocalDateTime needStartDay = LocalDateTime.now().plusDays(pullPreDay);

            if (localDate == null) { // 如果是第一次拉取，只拉取一定天数的数据
                startDay = needStartDay;
            }

            if (startDay.isBefore(needStartDay)) {
                startDay = needStartDay;
            }

            while(true) {
                List<ImSession> imSessionList = imSessionDao.getByUserIdAndTimeAndSize(userId, startDay, start, 200);
                LOGGER.info("start session, start: {}, size: {}", start, imSessionList.size());
                if (imSessionList.size() < 1) break;
                start += imSessionList.size();
                int sum = loadMessages(userId, imSessionList, startDay, isLatestChange);     // 同步消息
                size += sum;
            }

            long times = System.currentTimeMillis() - time;
            LOGGER.info("synData message total time: {}, total size {}", times, size);
            return size;
        });
    }

    /**
     * 同步消息事件
     * @param userId
     */
    public int synMessageAction(String userId) throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ImMessageDao imMessageDao = BeanFactory.getInstance().getBean(ImMessageDao.class);
        ImMessageSendDao imMessageSendDao = BeanFactory.getInstance().getBean(ImMessageSendDao.class);

        return synData(userId,ImSettingName.SynMessageAction.name(), (localDate, isLatestChange) -> {
            int size = 0;
            long time = System.currentTimeMillis();
            int pageSize = 1000;
            String serverMessageId = null;
            while(true) {
                try {
                    long startTime = System.currentTimeMillis();
                    MessagePullActionReq messagePullActionReq = new MessagePullActionReq();
                    messagePullActionReq.setAftDate(localDate);
                    messagePullActionReq.setServerMessageId(serverMessageId);
                    messagePullActionReq.setSize(pageSize);
                    MessageActionList messageActionList = rpcPullMessageService.pullMessageActionList(messagePullActionReq);

                    LOGGER.info("synData message action step1 time {}", System.currentTimeMillis() - time);

                    List<MessageActionItem> messageActionItemList = messageActionList.getMessageActionItemList();
                    size += messageActionItemList.size();
                    if (messageActionItemList.size() < 1) break;
                    LOGGER.info("synData message action size {}", messageActionList.getMessageActionItemList().size());

                    serverMessageId = messageActionItemList.get(messageActionItemList.size() - 1).getMessageId();
                    List<ImMessage> groupMessages = new ArrayList<>();

                    List<ImMessageActionData> p2PImMessageActionDataList = new ArrayList();
                    List<ImMessageActionData> groupImMessageActionDataList = new ArrayList();

                    Map<String, List<MessageActionItem>> messageActionItemMap = messageActionItemList.stream().collect(Collectors.groupingBy(MessageActionItem::getMessageId));
                    Set<String> serverIds = messageActionItemMap.keySet();
                    List<ImMessage> imMessages = imMessageDao.getByServerIdsAndUserId(serverIds, userId);
                    for (ImMessage imMessage : imMessages) {
                        List<MessageActionItem> messageActionItems = messageActionItemMap.get(imMessage.getServerId());
                        for (MessageActionItem messageActionItem :messageActionItems) {
                            ImMessageActionStatus actionStatus = ImMessageActionStatus.valueOf(messageActionItem.getMessageActionStatus());
                            if (imMessage.isP2p()) {
                                p2PImMessageActionDataList.add(new ImMessageActionData(imMessage.getId(), messageActionItem.getUserId(), actionStatus));
                            } else {
                                groupMessages.add(imMessage);
                                groupImMessageActionDataList.add(new ImMessageActionData(imMessage.getId(), messageActionItem.getUserId(), actionStatus));
                            }
                        }
                    }
                    LOGGER.info("synData message action step2 time {}", System.currentTimeMillis() - time);

                    if (p2PImMessageActionDataList.size() > 0) {
                        long st = System.currentTimeMillis();
                        imMessageDao.changeMessageActionStatus(p2PImMessageActionDataList);
                        long et = System.currentTimeMillis() - st;
                        LOGGER.info("synData message action change action status time {}", et);
                    }
                    if (groupImMessageActionDataList.size() > 0) {
                        long st = System.currentTimeMillis();
                        imMessageSendDao.changeMessageSendActionStatus(groupImMessageActionDataList);
                        long et = System.currentTimeMillis() - st;
                        LOGGER.info("synData message action change send action status time {}", et);
                    }

                    Set<Long> readCompleteMessageIdSet = new HashSet<>();
                    Set<Long> withdrawCompleteMessageIdSet = new HashSet<>();
                    for (ImMessage imMessage : groupMessages) {
                        boolean isSynEnd = imMessageSendDao.isSynEnd(imMessage.getId());
                        if (isSynEnd) {
                            if (imMessage.getActionStatus() == ImMessageActionStatus.read || imMessage.getActionStatus() == ImMessageActionStatus.readed) {
                                readCompleteMessageIdSet.add(imMessage.getId());
                            } else if (imMessage.getActionStatus() == ImMessageActionStatus.withdraw || imMessage.getActionStatus() == ImMessageActionStatus.withdrawed) {
                                withdrawCompleteMessageIdSet.add(imMessage.getId());
                            } else {
                                // do nothing
                                LOGGER.error("不支持这种状态{}", imMessage.getActionStatus());
                            }
                        }
                    }
                    LOGGER.info("synData message action step4 time {}", System.currentTimeMillis() - time);
                    if (readCompleteMessageIdSet.size() > 0) {
                        long st = System.currentTimeMillis();
                        imMessageDao.changeMessageActionStatus(readCompleteMessageIdSet, ImMessageActionStatus.readed);
                        long et = System.currentTimeMillis() - st;
                        LOGGER.info("synData message action change readed action status time {}", et);
                    }
                    if (withdrawCompleteMessageIdSet.size() > 0) {
                        long st = System.currentTimeMillis();
                        imMessageDao.changeMessageActionStatus(withdrawCompleteMessageIdSet, ImMessageActionStatus.withdrawed);
                        long et = System.currentTimeMillis() - st;
                        LOGGER.info("synData message action change withdrawed action status time {}", et);
                    }

                    LOGGER.info("synData message action step5 time {}", System.currentTimeMillis() - time);
                    MessagePullActionOkReq messagePullActionOkReq = new MessagePullActionOkReq();
                    List<MessageActionItem> okMessageActionItems = messageActionItemList.stream().filter(ma->ma.getType() == MessageActionType.owner).collect(Collectors.toList());
                    if (okMessageActionItems.size() > 0) {
                        List<MessageActionOkItem> messageActionItemRespList = okMessageActionItems.stream().map(mi->buildMessageActionOkItem(mi)).collect(Collectors.toList());
                        messagePullActionOkReq.setMessageActionOkItemList(messageActionItemRespList);

                        long endTime = System.currentTimeMillis();
                        OkResp okResp = rpcPullMessageService.okPullActionMessages(messagePullActionOkReq);
                        long time1 = System.currentTimeMillis() - endTime;
                        long time2 = System.currentTimeMillis() - startTime;

                        LOGGER.info("synData message action time1:{}, time2: {}", time1, time2);

                        if (!okResp.isSuccess()) {
                            LOGGER.error("消息事件确认失败!");
                        }
                    }

                    LOGGER.info("synData message action step6 time {}", System.currentTimeMillis() - time);
                    if (messageActionList.getMessageActionItemList().size() != pageSize) {
                        break;
                    }
                } catch (Exception e) {
                    LOGGER.error("同步消息错误", e);
                    throw new RuntimeException(e);
                }
            }
            long times = System.currentTimeMillis() - time;
            LOGGER.info("synData message action data time:" + times);
            return size;
        });
    }

    private <T>T synData(String userId, String name, SynBusiness<T> synBusiness) throws Exception {
        ImSettingDao imSettingDao = BeanFactory.getInstance().getBean(ImSettingDao.class);

        ImSetting imSetting = imSettingDao.getImSettingByUserIdAndName(userId, name);
        LocalDateTime localDate = null;
        boolean isLatestChange = false;  // 是否导入最近变更数据
        if (imSetting != null) { // 全量导入机构信息
            localDate = DateUtil.toLocalDateTime(imSetting.getValue(), "yyyy-MM-dd HH:mm");
            localDate = localDate.plusMinutes(-1);
            isLatestChange = true;
        }
        LOGGER.info("synData {} startTime {}", name, localDate);
        long time = System.currentTimeMillis();
        T t = null;
        try {
            t = synBusiness.doBusiness(localDate, isLatestChange);
            long endTime = System.currentTimeMillis() - time;
            LOGGER.info("synData {} time: {}, total: {}", name, endTime, t);
        } catch (Exception error) {
            LOGGER.error(String.format("synData %s error", name), error);
        }

        CacheService cacheService = RpcUtil.getRpcBootstrap().getRpcConfig().getCacheService();
        LocalDateTime serverDate = (LocalDateTime) cacheService.getCache("serverDate");
        if (isLatestChange) {
            imSetting.setValue(DateUtil.format(serverDate, "yyyy-MM-dd HH:mm"));
            imSetting.setUpdateTime(new Date());
            imSettingDao.updateImSetting(imSetting);
        } else {
            imSetting = new ImSetting();
            imSetting.setName(name);
            imSetting.setCurrentUserId(userId);
            imSetting.setValue(DateUtil.format(serverDate, "yyyy-MM-dd HH:mm"));
            imSetting.setUpdateTime(new Date());
            imSetting.setCreateTime(new Date());
            imSettingDao.insertImSetting(imSetting);
        }
        return t;
    }

    private int synConfigs(String userId) throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ConfigService configService = BeanFactory.getInstance().getBean(ConfigService.class);

        return synData("", ImSettingName.SynConfig.name(), (localDate, isLatestChange) -> {
            int total = 0;
            int pageSize = 500;
            ConfigPullReq configPullReq = new ConfigPullReq();
            configPullReq.setStartDate(localDate);
            configPullReq.setPageSize(pageSize);
            String startName = null;

            while(true) {
                configPullReq.setStartServerName(startName);
                ConfigItemList configItemList = rpcPullMessageService.pullConfigList(configPullReq);
                total += configItemList.getConfigItems().size();
                LOGGER.info("synData config size {}", configItemList.getConfigItems().size());

                int length = configItemList.getConfigItems().size();

                if (length > 0) {
                    int lastIndex = configItemList.getConfigItems().size() - 1;
                    startName = configItemList.getConfigItems().get(lastIndex).getName();
                    configService.updateOrInsert(userId, configItemList.getConfigItems());
                }

                if (length != pageSize) {
                    break;
                }
            }
            return total;
        });
    }

    private int synDepartments() throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ImDepartmentService imDepartmentService = BeanFactory.getInstance().getBean(ImDepartmentService.class);

        return synData("", ImSettingName.SynDepartment.name(), (localDate, isLatestChange) -> {
            int total = 0;
            int pageSize = 500;
            DepartmentInfoListReq departmentInfoListReq = new DepartmentInfoListReq();
            departmentInfoListReq.setLatestChange(isLatestChange);
            departmentInfoListReq.setStartDate(localDate);
            departmentInfoListReq.setPageSize(pageSize);
            String startOrgCode = null;

            while(true) {
                departmentInfoListReq.setStartOrgCode(startOrgCode);
                DepartmentInfoItemList departmentInfoItemList = rpcPullMessageService.pullDepartmentInfoList(departmentInfoListReq);
                total += departmentInfoItemList.getDepartmentInfoItemList().size();
                LOGGER.info("synData department size {}", departmentInfoItemList.getDepartmentInfoItemList().size());

                int length = departmentInfoItemList.getDepartmentInfoItemList().size();
                if (length > 0) {
                    int lastIndex = departmentInfoItemList.getDepartmentInfoItemList().size() - 1;
                    startOrgCode = departmentInfoItemList.getDepartmentInfoItemList().get(lastIndex).getOrgCode();
                    imDepartmentService.saveOrUpdates(departmentInfoItemList.getDepartmentInfoItemList());
                }

                if (length != pageSize) {
                    break;
                }
            }
            return total;
        });
    }

    private int synUsers() throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ImUserService imUserService = BeanFactory.getInstance().getBean(ImUserService.class);

        return synData("", ImSettingName.SynUser.name(), (localDate, isLatestChange) -> {
            int total = 0;
            String startUserId = null;
            int pageSize = 500;
            UserInfoListReq userInfoListReq = new UserInfoListReq();
            userInfoListReq.setPageSize(pageSize);
            userInfoListReq.setStartUserId(startUserId);
            userInfoListReq.setStartDate(localDate);

            while(true) {
                userInfoListReq.setStartUserId(startUserId);
                UserInfoItemList userInfoItemList = rpcPullMessageService.pullUserInfoList(userInfoListReq);
                int size = userInfoItemList.getUserInfoItemList().size();
                total += size;

                if (size > 0) {
                    startUserId = userInfoItemList.getUserInfoItemList().get(size - 1).getUserId();
                    imUserService.saveOrUpdates(userInfoItemList.getUserInfoItemList());
                }

                if (size != pageSize) {
                    break;
                }
            }

            return total;
        });
    }

    private int synGroups(String userId) throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ImGroupService imGroupService = BeanFactory.getInstance().getBean(ImGroupService.class);

        return synData(userId, ImSettingName.SynGroup.name(), (startDate, isLatestChange) -> {
            int size = 0;
            int pageSize = 200;
            GroupInfoListReq groupInfoListReq = new GroupInfoListReq();
            groupInfoListReq.setGroupType(GroupType.chat);
            groupInfoListReq.setStartDate(startDate);
            groupInfoListReq.setPageSize(pageSize);

            String startServerId = null;
            while(true) {
                groupInfoListReq.setStartServerId(startServerId);

                GroupInfoItemList groupInfoListItem = rpcPullMessageService.pullGroupInfoList(groupInfoListReq);
                size += groupInfoListItem.getGroupInfoItemList().size();
                LOGGER.info("synData groups size {}, {}", groupInfoListItem.getGroupInfoItemList().size(),
                        JsonUtil.toJson(groupInfoListItem));

                imGroupService.saveOrUpdates(userId, groupInfoListItem.getGroupInfoItemList(), false);
                if (groupInfoListItem.getGroupInfoItemList().size() == pageSize) {
                    startServerId = groupInfoListItem.getGroupInfoItemList().get(groupInfoListItem.getGroupInfoItemList().size() - 1).getGroupId();
                } else {
                    break;
                }
            }
            return size;
        });
    }

    private int synModules(String userId) throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ModuleService moduleService = BeanFactory.getInstance().getBean(ModuleService.class);

        return synData(userId, ImSettingName.SynModule.name(), (startDate, isLatestChange) -> {
            int size = 0;
            int pageSize = 100;
            ModulePullReq modulePullReq = new ModulePullReq();
            modulePullReq.setStartDate(startDate);
            modulePullReq.setPageSize(pageSize);

            String startServerName = null;
            while(true) {
                modulePullReq.setStartServerName(startServerName);

                ModuleInfoItemList moduleInfoItemList = rpcPullMessageService.pullModuleInfoList(modulePullReq);
                size += moduleInfoItemList.getModuleInfoItems().size();
                LOGGER.info("synData module size {}", moduleInfoItemList.getModuleInfoItems().size());

                moduleService.saveOrUpdates(userId, moduleInfoItemList.getModuleInfoItems());
                if (moduleInfoItemList.getModuleInfoItems().size() == pageSize) {
                    startServerName = moduleInfoItemList.getModuleInfoItems().get(moduleInfoItemList.getModuleInfoItems().size() - 1).getName();
                } else {
                    break;
                }
            }
            return size;
        });
    }

    private int synLabels(String userId) throws Exception {
        RpcBaseService rpcBaseService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcBaseService.class);
        LabelService labelService = BeanFactory.getInstance().getBean(LabelService.class);

        return synData(userId,ImSettingName.SynLabel.name(), (startDate, isLatestChange) -> {
            int size = 0;
            int pageSize = 1000;
            LabelPullReq labelPullReq = new LabelPullReq();
            labelPullReq.setStartDate(startDate);
            labelPullReq.setPageSize(pageSize);

            String startServerId = null;
            while(true) {
                labelPullReq.setServerStartId(startServerId);

                LabelInfoItemList labelListItem = rpcBaseService.pullLabels(labelPullReq);
                size += labelListItem.getLabelInfoItems().size();
                LOGGER.info("synData labels size {}", labelListItem.getLabelInfoItems().size());

                labelService.updateOrInsert(userId, labelListItem.getLabelInfoItems());

                Set<String> serverIdSet = labelListItem.getLabelInfoItems().stream().map(labelInfoItem -> labelInfoItem.getServerId()).collect(Collectors.toSet());
                OkListResp okListResp = new OkListResp(serverIdSet);
                rpcBaseService.pullLabelsResp(okListResp);

                if (labelListItem.getLabelInfoItems().size() != pageSize) {
                    break;
                }

                if(labelListItem.getLabelInfoItems().size() > 0) {
                    startServerId = labelListItem.getLabelInfoItems().get(0).getServerId();
                }
            }
            return size;
        });
    }

    private int synCollects(String userId) throws Exception {
        MessageRefService messageRefService = BeanFactory.getInstance().getBean(MessageRefService.class);
        CollectService collectService = BeanFactory.getInstance().getBean(CollectService.class);
        RpcBaseService rpcBaseService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcBaseService.class);
        SaveMessageService saveMessageService = BeanFactory.getInstance().getBean(SaveMessageService.class);

        return synData(userId,ImSettingName.SynCollect.name(), (startDate, isLatestChange) -> {
            int size = 0;
            int pageSize = 500;
            CollectPullReq collectPullReq = new CollectPullReq();
            collectPullReq.setStartDate(startDate);
            collectPullReq.setPageSize(pageSize);

            String startServerId = null;
            while(true) {
                collectPullReq.setServerStartId(startServerId);

                CollectItemList collectItemList = rpcBaseService.pullCollects(collectPullReq);
                size += collectItemList.getCollectInfoItems().size();
                LOGGER.info("synData collects size {}", collectItemList.getCollectInfoItems().size());

                // 如果消息有一引用，需要加载应用的消息
                List<MessageObject> refMessageObject = messageRefService.pullCollectRefMessages(userId, collectItemList.getCollectInfoItems());
                saveMessageService.saveMessages(userId, refMessageObject, true);

                collectService.doCollectBusiness(userId, collectItemList.getCollectInfoItems());

                Set<String> serverIdSet = collectItemList.getCollectInfoItems().stream().map(collectInfoItem -> collectInfoItem.getServerId()).collect(Collectors.toSet());
                OkListResp okListResp = new OkListResp(serverIdSet);
                rpcBaseService.pullCollectsResp(okListResp);

                if (collectItemList.getCollectInfoItems().size() != pageSize) {
                    break;
                }

                if(collectItemList.getCollectInfoItems().size() > 0) {
                    startServerId = collectItemList.getCollectInfoItems().get(0).getServerId();
                }
            }
            return size;
        });
    }

    private int synFeedbacks(String userId) throws Exception {
        RpcBaseService rpcBaseService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcBaseService.class);
        FeedbackService feedbackService = BeanFactory.getInstance().getBean(FeedbackService.class);

        return synData(userId,ImSettingName.SynFeedback.name(), (startDate, isLatestChange) -> {
            int size = 0;
            int pageSize = 500;
            FeedbackPullReq feedbackPullReq = new FeedbackPullReq();
            feedbackPullReq.setStartDate(startDate);
            feedbackPullReq.setPageSize(pageSize);

            String startServerId = null;
            while(true) {
                feedbackPullReq.setServerStartId(startServerId);

                FeedbackItemList feedbackItemList = rpcBaseService.pullFeedbacks(feedbackPullReq);
                size += feedbackItemList.getFeedbackItemList().size();
                LOGGER.info("synData feedbacks size {}", feedbackItemList.getFeedbackItemList().size());

                feedbackService.doPullFeedbackBusiness(userId, feedbackItemList.getFeedbackItemList());

                Set<String> serverIdSet = feedbackItemList.getFeedbackItemList().stream().map(feedbackItem -> feedbackItem.getServerId()).collect(Collectors.toSet());
                OkListResp okListResp = new OkListResp(serverIdSet);
                rpcBaseService.pullCollectsResp(okListResp);

                if (feedbackItemList.getFeedbackItemList().size() != pageSize) {
                    break;
                }

                if(feedbackItemList.getFeedbackItemList().size() > 0) {
                    startServerId = feedbackItemList.getFeedbackItemList().get(0).getServerId();
                }

            }
            return size;
        });
    }

    private int synSensitiveWords(String userId) throws Exception {
        RpcBaseService rpcBaseService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcBaseService.class);
        SensitiveWordService sensitiveWordService = BeanFactory.getInstance().getBean(SensitiveWordService.class);

        return synData("", ImSettingName.SynSensitiveWord.name(), (startDate, isLatestChange) -> {
            int size = 0;
            int pageSize = 500;
            SensitiveWordPullReq sensitiveWordPullReq = new SensitiveWordPullReq();
            sensitiveWordPullReq.setStartDate(startDate);
            sensitiveWordPullReq.setPageSize(pageSize);

            String startServerId = null;
            while(true) {
                sensitiveWordPullReq.setServerStartId(startServerId);

                SensitiveWordItemList sensitiveWordListItem = rpcBaseService.pullSensitiveWords(sensitiveWordPullReq);
                size += sensitiveWordListItem.getSensitiveWordList().size();
                LOGGER.info("synData sensitiveWords size {}", sensitiveWordListItem.getSensitiveWordList().size());

                sensitiveWordService.saveOrUpdates(userId, sensitiveWordListItem.getSensitiveWordList());

                if (sensitiveWordListItem.getSensitiveWordList().size() != pageSize) {
                    break;
                }

                if(sensitiveWordListItem.getSensitiveWordList().size() > 0) {
                    startServerId = sensitiveWordListItem.getSensitiveWordList().get(0).getServerId();
                }
            }
            return size;
        });
    }

    // 发送未成功的群
    private int sendNeedGroups(String userId) throws SQLException {
        ImGroupService imGroupService = BeanFactory.getInstance().getBean(ImGroupService.class);
        ImGroupDao imGroupDao = BeanFactory.getInstance().getBean(ImGroupDao.class);
        RpcGroupService rpcGroupService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcGroupService.class);

        int size = 0;
        int pageSize = 1000;

        while(true) {
            List<ImGroup> imGroups = imGroupDao.getNeedSendGroupsByUserId(userId, pageSize);
            size += imGroups.size();
            LOGGER.info("synData sendNeedGroups size {}", imGroups.size());
            for(ImGroup imGroup : imGroups) {
                GroupCreateReq groupCreateReq = new GroupCreateReq();
                groupCreateReq.setGroupName(imGroup.getName());
                groupCreateReq.setGroupId(imGroup.getCode());
                groupCreateReq.setGroupDesc(imGroup.getDescription());
                groupCreateReq.setUserIdSet(imGroup.getGroupUserSet().stream().map(u->u.getUserId()).collect(Collectors.toSet()));
                OkResp okResp = rpcGroupService.groupCreate(groupCreateReq);

                imGroupService.sendSuccess(imGroup, okResp.getOtherInfo());
            }
            if (imGroups.size() != pageSize) {
                break;
            }
        }

        return size;
    }

    // 发送未发送成功的消息
    private int sendNeedMessages(String userId) throws Exception {
        int size = 0;
        size += sendNeedCommonMessages(userId);   // 发送未成功的消息
        size += sendNeedFileMessages(userId);     // 发送上传未成功附件
        return size;
    }

    // 再次发送未发送成功的消息
    private int sendNeedCommonMessages(String userId) throws Exception {
        ImMessageDao imMessageDao = BeanFactory.getInstance().getBean(ImMessageDao.class);
        MessageSendService messageSendService = BeanFactory.getInstance().getBean(MessageSendService.class);

        int size = 0;
        int pageSize = 300;
        List<ImMessage> imMessages = imMessageDao.getNeedSendMessagesByUserId(userId, pageSize);
        size += imMessages.size();
        if (imMessages.size() < 1) return size;
        LOGGER.info("synData sendMessage size {}", imMessages.size());
        messageSendService.sendMessages(imMessages);
        if (imMessages.size() ==  pageSize) {
            size += sendNeedCommonMessages(userId);
        }
        return size;
    }

    /**
     * 发送上传未成功的附件
     */
    private int sendNeedFileMessages(String userId) throws SQLException {
        ImMessageDao imMessageDao = BeanFactory.getInstance().getBean(ImMessageDao.class);

        int size = 0;
        int pageSize = 1000;
        List<ImMessage> imMessages = imMessageDao.getNeedSendFileMessagesByUserId(userId, pageSize);
        size += imMessages.size();
        LOGGER.info("synData sendFileMessage size {}", imMessages.size());
        for(ImMessage imMessage : imMessages) {
            String serverId = imMessage.getServerId();
            sendAttachFile(serverId, imMessage);
        }
        return size;
    }

    /**
     * 发送未同步成功的消息事件
     * @param userId
     */
    private int sendMessageActionStatus(String userId) throws SQLException {
        ImMessageDao imMessageDao = BeanFactory.getInstance().getBean(ImMessageDao.class);

        int size = 0;
        int pageSize = MessageReadService.BATCH_SIZE;

        while(true) {
            List<ImMessage> imMessages = imMessageDao.getNeedSynActionStatusMessagesByUserId(userId, pageSize);
            size += imMessages.size();
            LOGGER.info("synData sendActonStatusMessage size {}", imMessages.size());
            List<ImMessage> messageReadMessages = new ArrayList<>();
            List<ImMessage> messageWithdrawMessages = new ArrayList<>();
            for(ImMessage message : imMessages) {
                if (message.getActionStatus() == ImMessageActionStatus.read) {
                    messageReadMessages.add(message);
                } else if (message.getActionStatus() == ImMessageActionStatus.withdraw) {
                    messageWithdrawMessages.add(message);
                } else {
                    LOGGER.error("同步消息事件不支持这种类型{}", message.getActionStatus().name());
                }
            }

            sendReadMessages(messageReadMessages);
            sendWithdrawMessages(messageWithdrawMessages);
            if (imMessages.size() != pageSize) {
                break;
            }

        }
        return size;
    }

    public int synImSessions(String userId) throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ImSessionService imSessionService = BeanFactory.getInstance().getBean(ImSessionService.class);

        return synData(userId,ImSettingName.SynSession.name(), (startDate, isLatestChange) -> {
            int size = 0;
            int pageSize = 500;
            SessionPullReq sessionPullReq = new SessionPullReq();
            sessionPullReq.setPageSize(pageSize);
            sessionPullReq.setStartDate(startDate);
            sessionPullReq.setSessionType(SessionType.chat);

            String startServerId = null;
            while(true) {
                sessionPullReq.setStartServerId(startServerId);
                SessionItemList sessionListItem = rpcPullMessageService.pullSessionList(sessionPullReq);
                size += sessionListItem.getSessionItemList().size();
                LOGGER.info("synData session size {}", sessionListItem.getSessionItemList().size());
                imSessionService.saveOrUpdates(userId, sessionListItem.getSessionItemList());

                List<SessionItem> sessionItems = sessionListItem.getSessionItemList();
                if (sessionItems.size() != pageSize) {
                    break;
                }
                if (sessionItems.size() > 0) {
                    startServerId = sessionItems.get(sessionItems.size() - 1).getServerId();
                }
            }
            return size;
        });
    }

    private void loadMessagesThread(List<ImSession> messageGroupList, LocalDateTime startDate, boolean isLatestChange) throws Exception {
        if (messageGroupList.size() < 1) return;
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        try {
            CountDownLatch countDownLatch = new CountDownLatch(messageGroupList.size());
            for(ImSession messageGroup : messageGroupList) {
                executorService.submit(()->{
                    try {
                        long time = System.currentTimeMillis();
                        int size = loadMessage(messageGroup, startDate, isLatestChange);
                        long last = System.currentTimeMillis() - time;
                        LOGGER.info("load message group message {} time : {}", size, last);
                    } catch (Exception e) {
                        LOGGER.error("同步消息错误2", e);
                    }
                    countDownLatch.countDown();
                });
            }
            countDownLatch.await();
        } finally {
            executorService.shutdown();
        }
    }

    private int loadMessages(String userId, List<ImSession> messageSessions, LocalDateTime startDate, boolean isLatestChange) throws Exception {
        if (messageSessions.size() < 1) return 0;

        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        RpcMessageService rpcMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcMessageService.class);
        MessageRefService messageRefService = BeanFactory.getInstance().getBean(MessageRefService.class);
        SaveMessageService saveMessageService = BeanFactory.getInstance().getBean(SaveMessageService.class);

        int total = 0;
        int pageSize = 200;
        if (startDate != null) pageSize = 300;
        MessagePullBySessionIdsReq messagePullBySessionIdsReq = new MessagePullBySessionIdsReq();
        Set<String> sessionIds = messageSessions.stream().filter(ms->ms.getServerId() != null).map(ms->ms.getServerId()).collect(Collectors.toSet());
        if (sessionIds.size() < 1) return 0;
        messagePullBySessionIdsReq.setSessionIds(sessionIds);

        messagePullBySessionIdsReq.setTerminalType(TerminalType.pc);
        messagePullBySessionIdsReq.setAftDate(startDate);
        messagePullBySessionIdsReq.setPageSize(pageSize);

        String serverMessageId = null;
        while(true) {
            long time = System.currentTimeMillis();
            messagePullBySessionIdsReq.setStartServerMessageId(serverMessageId);
            MessageObjectList messageObjectList = rpcPullMessageService.pullMessageBySessionIds(messagePullBySessionIdsReq);
            if(messageObjectList.getMessageObjectList().size() < 1) break;

            total += messageObjectList.getMessageObjectList().size();
            long last = System.currentTimeMillis() - time;
            LOGGER.info("synData message size {}, time : {}", messageObjectList.getMessageObjectList().size(), last);

            Set<String> pulledServerIds = messageObjectList.getMessageObjectList().stream().map(mo->mo.getServerId()).collect(Collectors.toSet());
            Set<String> serverIdSet = messageObjectList.getMessageObjectList().stream().filter(m->m.isRef())
                    .flatMap(m-> Arrays.stream(m.getRefMessageIds().split(","))).filter(serverId->!pulledServerIds.contains(serverId)).collect(Collectors.toSet());
            List<MessageObject> messageObjects = messageRefService.pullRefMessages(userId, serverIdSet.stream().collect(Collectors.toList()));

            List<MessageObject> allMessageObjects = messageObjectList.getMessageObjectList();
            allMessageObjects.addAll(messageObjects);
            List<ImMessage> imMessageList = saveMessageService.saveMessages(userId, allMessageObjects, true);

            LOGGER.info("synData message save message end!! size {}", imMessageList.size());
            downloadFile(imMessageList);

            OkListReq okListReq = new OkListReq(messageObjectList.getMessageObjectList().stream().map(m->m.getLocalSeq()).collect(Collectors.toSet()));
            okListReq.setCoder((short)11);
            rpcMessageService.okMessages(okListReq);

            if (messageObjectList.getMessageObjectList().size() == pageSize) {
                serverMessageId = messageObjectList.getMessageObjectList().get(messageObjectList.getMessageObjectList().size()-1).getServerId();
            } else {
                break;
            }
        }
        return total;
    }

    private void downloadFile(List<ImMessage> imMessages) {
        for (ImMessage message : imMessages) {
            if(message.getFileStatus() != ImMessageFileStatus.downloaded && message.getMessageType().isFile()) {
                RpcUtil.getImService().downloadFile(message, new DefaultFileProgressListener(), m-> UiUtil.uiUpdateMessage(m));
            }
        }
    }

    private int loadMessage(ImSession messageSession, LocalDateTime startDate, boolean isLatestChange) throws Exception {
        RpcPullMessageService rpcPullMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcPullMessageService.class);
        ImGroupDao imGroupDao = BeanFactory.getInstance().getBean(ImGroupDao.class);
        RpcMessageService rpcMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcMessageService.class);
        SaveMessageService saveMessageService = BeanFactory.getInstance().getBean(SaveMessageService.class);

        int pageSize = 100;
        if (startDate != null) pageSize = 200;
        CacheService cacheService = RpcUtil.getRpcBootstrap().getRpcConfig().getCacheService();
        String userId = (String) cacheService.getCache(ImConfig.LOGIN_USER_ID);
        MessagePullByUserOrGroupReq messagePullByUserOrGroupReq = new MessagePullByUserOrGroupReq();
        if (messageSession.getGroupCode() != null) {
            ImGroup imGroup = imGroupDao.getByUserIdAndId(messageSession.getCurrentUserId(), messageSession.getGroupCode());
            messagePullByUserOrGroupReq.setGroupId(imGroup.getServerId());
            return 1;
        } else {
            messagePullByUserOrGroupReq.setFromUserId(messageSession.getFromUserId());
            if (userId.equals(messageSession.getFromUserId())) {
                messagePullByUserOrGroupReq.setFromUserId(messageSession.getToUserId());
            }
        }
        messagePullByUserOrGroupReq.setTerminalType(TerminalType.pc);
        messagePullByUserOrGroupReq.setAftDate(startDate);
        messagePullByUserOrGroupReq.setPageSize(pageSize);
        long time = System.currentTimeMillis();
        MessageObjectList messageListItem = rpcPullMessageService.pullMessageByMessageId(messagePullByUserOrGroupReq);
        long last = System.currentTimeMillis() - time;
        LOGGER.info("synData message time : {}", last);
        List<ImMessage> imMessageList = saveMessageService.saveMessages(userId, messageListItem.getMessageObjectList(), true);
        downloadFile(imMessageList);

        rpcMessageService.okMessages(new OkListReq(messageListItem.getMessageObjectList().stream().map(m->m.getLocalSeq()).collect(Collectors.toSet())));
//            LOGGER.info("syn message " + serverRedirectMessages.getContent().size() + " " + JsonUtil.toJson(messageGroup) + " time:" + (System.currentTimeMillis() - time1));
        return messageListItem.getMessageObjectList().size();
    }

    private void sendAttachFile(String serverId, ImMessage imMessage) {
        if (imMessage.getMessageType().isFile()) {
            try {
                RpcUtil.getImService().sendFile(imMessage, new DefaultFileProgressListener());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void sendReadMessages(List<ImMessage> imMessages) throws SQLException {
        if (imMessages.size() < 1) return;
        RpcMessageService rpcMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcMessageService.class);
        ImMessageDao imMessageDao = BeanFactory.getInstance().getBean(ImMessageDao.class);

        MessageReadReq messageReadReq = new MessageReadReq();
        Set<String> messageReadIds = imMessages.stream().map(m->m.getServerId()).collect(Collectors.toSet());
        messageReadReq.setMessageIds(messageReadIds);
        OkResp okResp = rpcMessageService.sendReadMessages(messageReadReq);
        Set<Long> readActionSet = new HashSet<>();
        if (okResp.isSuccess()) {
            for (ImMessage message : imMessages) {
                readActionSet.add(message.getId());
            }
            imMessageDao.changeMessageActionStatus(readActionSet, ImMessageActionStatus.readed);
        }
    }

    private void sendWithdrawMessages(List<ImMessage> imMessages) throws SQLException {
        if (imMessages.size() < 1) return;
        RpcMessageService rpcMessageService = RpcUtil.getRpcBootstrap().getRpcConfig().getClientBean(RpcMessageService.class);
        ImMessageDao imMessageDao = BeanFactory.getInstance().getBean(ImMessageDao.class);

        MessageWithdrawReq messageWithdrawReq = new MessageWithdrawReq();

        Set<String> messageWithdrawIds = imMessages.stream().map(m->m.getServerId()).collect(Collectors.toSet());
        messageWithdrawReq.setMessageIds(messageWithdrawIds);
        OkResp okResp = rpcMessageService.sendWithdrawMessages(messageWithdrawReq);
        Set<Long> withdrawActionSet = new HashSet<>();
        if (okResp.isSuccess()) {
            for (ImMessage message : imMessages) {
                if (message.getActionStatus() == ImMessageActionStatus.withdraw) {
                    withdrawActionSet.add(message.getId());
                }
            }
            imMessageDao.changeMessageActionStatus(withdrawActionSet, ImMessageActionStatus.withdrawed);
        }
    }

    private MessageActionOkItem buildMessageActionOkItem(MessageActionItem mi) {
        MessageActionOkItem messageActionOkItem = new MessageActionOkItem();
        messageActionOkItem.setMessageId(mi.getMessageId());
        messageActionOkItem.setUserId(mi.getUserId());
        return messageActionOkItem;
    }

    @FunctionalInterface
    private interface SynBusiness<T> {
        T doBusiness(LocalDateTime startDate, boolean isLatestChange) throws Exception;
    }
}
